package com.bonc.function;

import com.bonc.entity.BaowenEntity;
import org.apache.http.Consts;
import org.apache.http.HttpEntity;
import org.apache.http.NameValuePair;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.util.*;

public class GetSensorValue implements Runnable{
    static Properties prop=new Properties();
    static InputStream is=Thread.currentThread().getContextClassLoader().getSystemResourceAsStream("sword.properties");

    public static void getSensor() throws IOException, SQLException, InterruptedException  {
        prop.load(is);
        /**
         * 这里建立kafka连接，之建立一次
         */
        Properties props = new Properties();
        props.put("bootstrap.servers" ,prop.getProperty("brokerList"));
        props.put("key.serializer" ,"org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer" ,"org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks" ,"1");
        //props.put("connections.max.idle.ms" ,"10000");
        KafkaProducer producer = new KafkaProducer<>(props);
        producer = new KafkaProducer<String, String>(props);


        String ADD_URL=prop.getProperty("lingyuurl");
        String topic =prop.getProperty("topic");

        while (true){
            ArrayList<BaowenEntity> baos = GetInfoFromsql("180");
            for (BaowenEntity bao:baos){
                try{
                    Map<String, String> map = new HashMap<String, String >();
                    String oneKey=bao.getGatewayCode()+":"+bao.getDeviceCode();
           /* map.put("type", "1");
            map.put("deviceCode", "80 9A");
            map.put("gatewayCode", "00 00 F0 0B");
            map.put("address", "00");*/
                    map.put("type",bao.getType().toString());
                    map.put("deviceCode",bao.getDeviceCode());
                    map.put("gatewayCode",bao.getGatewayCode());
                    map.put("address",bao.getAddress());

                    String ret = sendPost(ADD_URL, map);

                    switch (bao.getSensorType()){
                        case "TEMPERATURE":
                            producer.send(new ProducerRecord<String, String>(topic,"TE:"+oneKey,ret));//温度
                            System.out.println("temperature"+ret);
                            break;
                        case "HUMIDITY":
                            producer.send(new ProducerRecord<String, String>(topic,"HU:"+oneKey,ret));//温度
                            System.out.println("humidity"+ret);
                            break;
                        case "NH3":
                            producer.send(new ProducerRecord<String, String>(topic,"NH:"+oneKey,ret));//氨气
                            System.out.println("NH3"+ret);
                            break;
                        case "PM25":
                            producer.send(new ProducerRecord<String, String>(topic,"PM:"+oneKey,ret));//pm25
                            System.out.println("PM25"+ret);
                            break;
                    }
                    //{"data":"15.5","gatewayStatus":"1","deviceStatus":"1","acquisitionTime":"2019-01-20T02:35:13.000+0000","success":"0","message":"查询成功!"}
                }
                catch (Exception e){
                    e.printStackTrace();
                    System.out.println("凌宇平台数据接收出现错误，请检查数据库");
                    continue;
                }
            }
        Thread.sleep(900000);//每隔15分钟取一次
        }
    }
    public static String sendPost(String url, Map<String, String> map) {
        CloseableHttpClient httpclient = HttpClients.createDefault();
        List<NameValuePair> formparams = new ArrayList<NameValuePair>();
        for (Map.Entry<String, String> entry : map.entrySet()) {
            formparams.add(new BasicNameValuePair(entry.getKey(), entry.getValue()));
        }
        // 编码
        UrlEncodedFormEntity formEntity = new UrlEncodedFormEntity(formparams, Consts.UTF_8);
        // 取得HttpPost对象
        HttpPost httpPost = new HttpPost(url);
        // 防止被当成攻击添加的
        //httpPost.setHeader("User-Agent", "Mozilla/5.0 (Windows NT 6.2; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.87 Safari/537.36");
        // 参数放入Entity
        httpPost.setEntity(formEntity);
        CloseableHttpResponse response = null;
        String result = null;
        try {
            response = httpclient.execute(httpPost);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println(e);
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException e) {
                    e.getMessage();
                }
            }
        }
        return result;
    }
    public static ArrayList<BaowenEntity> GetInfoFromsql(String sensorType) throws SQLException, IOException {
        prop.load(is);
        Connection connection=DriverManager.getConnection(prop.getProperty("jdbc.url"),prop.getProperty("jdbc.user"), prop.getProperty("jdbc.password"));
        PreparedStatement ps=connection.prepareStatement("select l_type,l_deviceCode,l_gatewayCode,l_address,sensor_type from tbl_data_url where  bangding_type=1 and l_type is NOT NULL and bindingData  in (select distinct(bindingData) from tbl_databind where toilet_SeatCode is not null)");
        //上面加了个条件，用于取得绑定的传感器的名字，不绑定的不要
        ResultSet resultSet=ps.executeQuery();
        ArrayList<BaowenEntity> baowenEntities=new ArrayList<>();
        while (resultSet.next()){
            BaowenEntity baowenEntity=new BaowenEntity();
            baowenEntity.setType(resultSet.getInt(1));
            baowenEntity.setDeviceCode(resultSet.getString(2));
            baowenEntity.setGatewayCode(resultSet.getString(3));
            baowenEntity.setAddress(resultSet.getString(4));
            baowenEntity.setSensorType(resultSet.getString(5));
            baowenEntities.add(baowenEntity);
        }
        if (ps!=null)ps.close();
        if (connection!=null)connection.close();
        return baowenEntities;
    }

    public static void main(String[] args) throws IOException, SQLException, InterruptedException {
        GetSensorValue getSensorValue=new GetSensorValue();
        getSensorValue.run();//这里明天测试一下
    }

    @Override
    public void run() {
        try {
            getSensor();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
